package leo.peersim.paxos;

import leo.peersim.common.LogicalClock;

import org.apache.log4j.Logger;

import peersim.core.CommonState;

/**
 * Acceptor agent
 * 
 * @author leo
 * @version June 20, 2009
 */
public class Acceptor extends PaxosAgent {
	
	private Logger logger = Logger.getLogger(this.getClass().getName());
	
	private String name = "Acceptor";

	private LogicalClock _latestBallot;
	
	private AcceptedBallots _accepted = new AcceptedBallots();
	
	public Acceptor(Paxos container) {
		super(container);
	}
	
	protected String getName() {
		return this.name;
	}
	
	/**
	 * 
	 * @param p
	 */
	protected void manage(Prepare p) {
		if (!container.updateLatestBallot(p.getBallotId())) {
			//NACK
			logger.trace(pp() + "can't accept prepare " + p.getBallotId() + " since I promised already " + container.getLatestPhase1Ballot());
			Refuse ref = new Refuse(CommonState.getNode(), container.getLogicalClock(), p.getBallotId(), container.getLatestPhase1Ballot());
			container.send(ref, p.getSenderNode());
			return;
		} else {
			//ACK
			container._estimateOfRecentLeader = p.getSenderNode();
			
			logger.trace(pp() + "received message! ACK");
			Promise promise = new Promise(CommonState.getNode(), container.getLogicalClock(), p.getBallotId(), container.getLastPhase2Ballot(), container.getLatestAcceptValue()+1);
			container.send(promise, p.getSenderNode());
		}
	}
	
	/**
	 * 
	 * @param a
	 */
	protected void manage(Accept a) {
		if (!container.checkLatestPhaseABallot(a.getBallotId())) {
			//NACK
			logger.debug(pp() + "can't accept " + a.getBallotId() + " since I promised not to accept less than " + container.getLatestPhase1Ballot());
			Reject rej = new Reject(CommonState.getNode(), container.getLogicalClock(), a.getBallotId(), container.getLatestPhase1Ballot() );
			container.send(rej, a.getSenderNode());
			return;
		} else {
			//ACK
			logger.trace(pp() + "received accept! ACK");
			container.updateLatestAcceptValue(a.getProposal());
			container.updateLatestPhase2Ballot(a.getBallotId());
			
			if (!a.getBallotId().equals(this._latestBallot)) {
				this._latestBallot = a.getBallotId();
				
				this._accepted = new AcceptedBallots();
				
				Accept accept = new Accept(CommonState.getNode(), container.getLogicalClock(), a.getBallotId(), a.getOpName(), a.getProposal());
				container.broadcastToGroup(accept);
			} else {
				if(!this._accepted.containsKey(a.getSenderNode())) {
					
					if(this._accepted.hasQuorum()) {
						//has already a quorum do nothing
						//TODO
					} else {
						
						this._accepted.put(a.getSenderNode(), a);
						if(this._accepted.hasQuorum()) {
							logger.trace(pp() + "quorum on ballot " + a.getBallotId() + " whit value _" + a.getProposal() + "_");
							//logger.info(pp() + "quorum size: " + this._accepted.size() );
							
							Decide dec = new Decide(CommonState.getNode(), container.getLogicalClock(), this._latestBallot, a.getOpName(), a.getProposal());
							//container.send(dec, CommonState.getNode());
							container.broadcastToGroup(dec);
						}
					}
				} else {
					// has already an accept for that node, dropping message
					//TODO
				}
			}
		}
	}
}
